package com.lsx143.wordcount.day3.fight;

import com.lsx143.wordcount.bean.AdsClickLog;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Test3_ADS {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
                .readTextFile("input/UserBehavior.csv")
                .map(new MapFunction<String, Tuple2<Tuple2<String, Long>, Long>>() {
                    @Override
                    public Tuple2<Tuple2<String, Long>, Long> map(String line) throws Exception {
                        String[] datas = line.split(",");
                        AdsClickLog adsClickLog = new AdsClickLog(
                                Long.valueOf(datas[0]),
                                Long.valueOf(datas[1]),
                                datas[2],
                                datas[3],
                                Long.valueOf(datas[4])
                        );
                        return new Tuple2<>(new Tuple2<>(adsClickLog.getProvince(), adsClickLog.getAdId()), 1L);
                    }
                })
                .keyBy(new KeySelector<Tuple2<Tuple2<String, Long>, Long>, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> getKey(Tuple2<Tuple2<String, Long>, Long> t) throws Exception {
                        return t.f0;
                    }
                })
                .sum(1)
                .print("省份广告");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}